
use chrono::{NaiveDateTime, Duration as ChronoDuration};
use crate::data_frame::value::{RealValue, ConfValue, ElementValue};
use crate::logical_plan::expr::window::{LogicalPlanExprWindow};
use crate::logical_plan::expr::func::LogicalPlanExprFuncOrValue;

pub mod partition_by;
pub mod aggregation;
pub mod accumulator;
pub mod window;
pub mod func;
pub mod general;

#[derive(Debug,Clone)]
pub enum Operator {
    Eq,
    NotEq,
    Lt,
    Le,
    Gt,
    Ge,
    And,
    Or,
}
impl Operator{
    pub fn to_func_name(&self)->&str{
        match self{
            Operator::Eq => {"eq"},
            Operator::NotEq => {"neq"},
            Operator::Lt => {"lt"},
            Operator::Le => {"le"},
            Operator::Gt => {"gt"},
            Operator::Ge => {"ge"},
            Operator::And => {"and"},
            Operator::Or => {"or"},
        }
    }
    pub fn new(v: &str)->Option<Self>{
        if v == "="{
            Some(Operator::Eq)
        }else if v == "!="{
            Some(Operator::NotEq)
        }else if v == "<"{
            Some(Operator::Lt)
        }else if v == "<="{
            Some(Operator::Le)
        }else if v == ">"{
            Some(Operator::Gt)
        }else if v == ">="{
            Some(Operator::Ge)
        }else if v == "&&"{
            Some(Operator::And)
        }else if v == "||"{
            Some(Operator::Or)
        }else{
            None
        }
    }
}

#[derive(Debug)]
pub enum LiteralValue{
    Null,
    Boolean(bool),
    Utf8(String),
    UInt8(u8),
    UInt16(u16),
    UInt32(u32),
    UInt64(u64),
    Int8(i8),
    Int16(i16),
    Int32(i32),
    Int64(i64),
    Float32(f32),
    Float64(f64),
    Range{
        left: u64,
        right: u64,
    },
    DateTime(NaiveDateTime),
    Duration(ChronoDuration),
}

#[derive(Debug)]
pub enum LogicalPlanExpr {
    OutSideExpr(String), //由于Logical支持复杂嵌套的定义方式，为了防止嵌套过深不便于阅读，允许预定义一些LogicalPlan来引用
                            // #<String># -> OutSideExpr
    ConfValue(ConfValue),//不指定值类型的配置值      String -> ConfValue
    RealValue(RealValue),//指定了值类型的配置值      <aaaaa> ->RealValue
    ElementValue(ElementValue),//某个元素的值       {a b c} -> ElementValue
    //>/</=/!=/>=/<= 等一系列两个元素的关系判断表达式集合
    //Binary(Box<LogicalPlanExpr>, Operator, Box<LogicalPlanExpr>),// <LogicalPlanExpr> op <LogicalPlanExpr>
    Func(LogicalPlanExprFuncOrValue),                 //可嵌套的函数 func_name(func_args...)
    /*
        Window{
            partition_by: disorder({src_ip},{src_port},{http,url})
            accumulator: and(count_gt(5),count_lt(10)
            window_time: 10,
            slide_time: 1,
            time_unit: sec,
        }
    */
    Window(LogicalPlanExprWindow),
    /*
    Take{
        expr: Box<LogicalPlanExpr>,
        idx: Box<LogicalPlanExpr>,
    },//根据idx取值
    SortBy{
        expr: Box<LogicalPlanExpr>,
        by: Vec<LogicalPlanExpr>,
        reverse: Vec<bool>,
    },
    Agg(AggExpr),
    Window{
        function: Box<LogicalPlanExpr>,
        partition_by: Vec<LogicalPlanExpr>,
        partition_type: PartitionType,
    },
    Wildcard,//通配符
    Slice{
        input: Box<LogicalPlanExpr>,
        offset: Box<LogicalPlanExpr>,
        length: Box<LogicalPlanExpr>,
    },
    Count,
    */
}

#[test]
pub fn test_parse_logical_expr(){
    let desc = "{a b c}";
    let logical_plan = parse_text::<LogicalPlanExpr,_>(desc,
                                                       parse_logical_expr_func(),
                                                       vec!["no"].as_slice());
    println!("logical_plan1: {:?}", logical_plan);

    let desc = "add_element({id},uuid(),<other_logical_expr>)";
    let logical_plan = parse_text::<LogicalPlanExpr,_>(desc,
                                                       parse_logical_expr_func(),
                                                       vec!["no"].as_slice());
    println!("logical_plan2: {:?}", logical_plan);
}
/*
fn parse_logical_expr_func<'a>() ->impl Parser<char, LogicalPlanExpr, TextChar>+'a{
    let parse_expr: Recursive<'a,char, LogicalPlanExpr, TextChar>= recursive::<'a,char,LogicalPlanExpr,TextChar,_,_>(|parse_expr: Recursive<'a,char,LogicalPlanExpr,TextChar>| {

        //let parse_conf_value = text_number().map(|o| Expr::Num(usize::from_str(o.as_str()).unwrap()));

        let parse_conf_value = text_key_world().map_end(|o| LogicalPlanExpr::ConfValue(ConfValue::new(o)));

        let parse_element_value = text_key_world()
                                                    .separated_by(text_just(" "),false)
                                                    .delimited_by(text_just("{"), text_just("}"))
                                                    .map_end(|o| LogicalPlanExpr::ElementValue(ElementValue::new(o)));

        let parse_func_no_arg = text_ident()
            .then_ignore(text_just("("))
            .then_ignore(text_just(")"))
            .map_end(|o| LogicalPlanExpr::Func(o, Vec::new()));

        let parse_outside_expr = text_ident()
                .delimited_by(text_just("#<"), text_just(">#"))
                .map_end(|o| LogicalPlanExpr::OutSideExpr(o));

        let parse_func_with_args = text_ident()
            .and(
                parse_expr.clone()
                    .separated_by(text_just(","), false)
                    .delimited_by(text_just("("),text_just(")"))
            )
            .map_end(|(n,f)| LogicalPlanExpr::Func(n, f));

        /*
        let op = text_just("=").padded_by(text_just(" "))
            .or(text_just("!=").padded_by(text_just(" ")))
            .or(text_just("<").padded_by(text_just(" ")))
            .or(text_just("<=").padded_by(text_just(" ")))
            .or(text_just(">").padded_by(text_just(" ")))
            .or(text_just(">=").padded_by(text_just(" ")))
            .or(text_just("&&").padded_by(text_just(" ")))
            .or(text_just("||").padded_by(text_just(" ")));

        let parse_binary_expr = parse_expr.clone().delimited_by(text_just("<"), text_just(">"))
            .and(
                op
            )
            .and(
                parse_expr.clone().delimited_by(text_just("<"), text_just(">"))
            )
            .map_end(|((left,op),right)| LogicalPlanExpr::Binary(Box::new(left),Operator::new(op.as_str()).unwrap(),Box::new(right)) );
        */
        let atom = parse_conf_value
                                        .or(parse_outside_expr)
                                        .or(parse_element_value)
                                        .or(parse_func_no_arg)
                                        .or(parse_func_with_args)
                                        ;

        atom
    });
    parse_expr
}

 */

pub enum ApplyOptions{
    ApplyGroups,//不太明白跟下面的区别
    ApplyList,//把groups数据收集成list，再调用函数
    ApplyFlat,//在收集之前先调用函数
}
pub struct FunctionOptions{
    pub collect_groups: ApplyOptions,
    pub input_wildcard_expansion: bool,//输入的通配符是否展开
}

impl LogicalPlanExpr {
    //基于用户配置来构建LogicalPlanExpr，而不是其他LogicalPlanExpr
    /*
    pub fn col(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            let t = &args[0];
            if let FuncArg::Column(t) = t{
                return MapResult::Ok(LogicalPlanExpr::Column(ElementValue{element:t.clone()}));
            }else{
                return MapResult::Err(ResultErr::new(format!("col only accept FuncArg::Column")));
            }
        }else if args.len() > 1{
            let mut tmp = Vec::new();
            for t in args.iter(){
                if let FuncArg::Column(t) = t{
                    tmp.push(ElementValue{element:t.clone()});
                }else{
                    return MapResult::Err(ResultErr::new(format!("col only accept FuncArg::Column")));
                }
            }
            return MapResult::Ok(LogicalPlanExpr::Columns(tmp));
        }else{
            return MapResult::Err(ResultErr::new(format!("col need at least one arg")));
        }
    }
    pub fn literal(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            let t = &args[0];
            if let FuncArg::Literal(t) = t{
                return MapResult::Ok(LogicalPlanExpr::LiteralConf(ConfValue::new(t.to_string())));
            }else{
                return MapResult::Err(ResultErr::new(format!("literal only accept FuncArg::Literal")));
            }
        }else{
            return MapResult::Err(ResultErr::new(format!("literal need at least one arg")));
        }
    }

    //基于已有LogicalPlanExpr来构建复杂LogicalPlanExpr
    pub fn eq(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("eq left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("eq right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Eq, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("eq args len must eq 2, {:?}", args)));
        }
    }
    pub fn neq(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("neq left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("neq right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::NotEq, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("neq args len must eq 2, {:?}", args)));
        }
    }
    pub fn lt(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("lt left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("lt right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Lt, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("lt args len must eq 2, {:?}", args)));
        }
    }
    pub fn le(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("le left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("le right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Le, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("le args len must eq 2, {:?}", args)));
        }
    }
    pub fn gt(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("gt left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("gt right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Gt, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("gt args len must eq 2, {:?}", args)));
        }
    }
    pub fn ge(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("ge left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("ge right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Ge, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("ge args len must eq 2, {:?}", args)));
        }
    }
    pub fn not(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            if let FuncArg::OtherExpr(left) = &args[0]{
                if let Some(left) = pre_expr.get(left){
                    return MapResult::Ok(LogicalPlanExpr::Not(Box::new(left.clone())));
                }else{
                    return MapResult::Err(ResultErr::new(format!("not left Expr not exist")));
                }
            }else{
                return MapResult::Err(ResultErr::new(format!("not left must be Expr")));
            }
        }else{
            return MapResult::Err(ResultErr::new(format!("not args len must eq 1")));
        }
    }
    pub fn and(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("and left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("and right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::And, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("and args len must eq 2, {:?}", args)));
        }
    }
    pub fn or(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("or left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("or right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Or, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("or args len must eq 2, {:?}", args)));
        }
    }

     */
}

/*
三大类表达式：
1. filter类
2. 简单处理/判断函数类型
3. 复杂处理类

conf支持层层嵌套的方式来配置 LogicalPlanExpr 表达式:

filter类：
    支持((<a> and <b>) || (<c> or <d>)) && <e>复杂灵活风格的配置
简单函数类：
    优先从LogicalPlanExpr方法中去匹配函数，如果没找到，就调用logicalplan_func_map_physicalplan
    从PhysicalPlan里面去找函数。都没找到就是不支持。

column("port") column_port
lit("80")      port_value_80
gt(<column_port>,<port_value_80>)  my_gt
*/

/*
pub fn conf_map_to_logicalplan(conf: &str)->MapResult<LogicalPlanExpr>{

    info!("conf_map_to_logicalplan, conf: {:?}", conf);

    let logical_plan = parse_text::<LogicalPlanExpr,_>(conf,
                                                       &parse_logical_expr_func(),
                                                            vec!["no"].as_slice());
    match logical_plan{
        ParserResult::OkEnd(lp) => {
            MapResult::Ok(lp)
        },
        ParserResult::Ok(_) => {
            MapResult::Err(MapErr::new(format!("Ok not End")))
        },
        ParserResult::Err(e) => {
            MapResult::Err(MapErr::new(format!("Err at {}", e.offset)))
        },
        ParserResult::ErrEnd(e) => {
            MapResult::Err(MapErr::new(format!("ErrEnd at {}", e.offset)))
        },
    }
}

 */










